package io.modelcontextprotocol.sdk;

import io.modelcontextprotocol.client.McpClient;
import io.modelcontextprotocol.client.McpSyncClient;
import io.modelcontextprotocol.client.transport.HttpClientSseClientTransport;
import io.modelcontextprotocol.json.McpJsonMapper;
import io.modelcontextprotocol.spec.McpClientTransport;
import io.modelcontextprotocol.spec.McpSchema;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.time.Duration;
import java.util.Map;
import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicInteger;

/**
 * 聊天MCP客户端测试 - 演示如何使用聊天MCP服务
 * <p>
 * 这个测试展示了聊天MCP服务的核心功能： 1. 获取聊天记录 2. 发送消息 3. 实时日志推送 4. 错误处理 5. Ping/Pong断线重连
 */
public class ChatMcpClientTest {

    private static final Logger logger = LoggerFactory.getLogger(ChatMcpClientTest.class);

    // 配置模拟器转发,把模拟器宿主机的ip端口映射的模拟器的8081端口上: adb forward tcp:8081 tcp:8081, 在宿主机上访问: http://localhost:8081 就可以连接到模拟器了
    private static final String SERVER_URL = "http://localhost:8081";
//    private static final String SERVER_URL = "http://192.168.0.104:8081";
    
    // 是否使用自定义会话ID（设置为null则使用服务器生成的UUID）
    private String CUSTOM_SESSION_ID = "client-" + "00001"; // 这里可以修改为不同的ID来模拟多个客户端连接

    // Ping/Pong 配置
    private static final long PING_INTERVAL_SECONDS = 30; // ping间隔时间
    private static final long PING_TIMEOUT_SECONDS = 5;   // ping超时时间
    private static final int MAX_RECONNECT_ATTEMPTS = 5;  // 最大重连次数
    private static final long RECONNECT_DELAY_SECONDS = 3; // 重连延迟时间

    public static void main(String[] args) {
        new ChatMcpClientTest().start();
    }

    public  void start() {
        logger.info("🚀 聊天MCP客户端测试 - 演示聊天服务功能（支持断线重连）");
        logger.info("==================================================");
        logger.info(SERVER_URL + "/sse");
        logger.info(SERVER_URL + "/message");

        CUSTOM_SESSION_ID = CUSTOM_SESSION_ID + "-" + System.currentTimeMillis();

        // 创建带有断线重连功能的客户端管理器
        ReconnectableMcpClient reconnectableClient = new ReconnectableMcpClient(SERVER_URL);
        
        try {
            // 启动客户端（包含ping/pong心跳检测）
            reconnectableClient.start();
            
            logger.info("✅ 客户端已启动，心跳检测已开启");
            
            // 等待连接成功（最多等待60秒）
            if (!reconnectableClient.isConnected()) {
                logger.info("⏳ 等待客户端连接到服务器...");
                int waitCount = 0;
                while (!reconnectableClient.isConnected() && waitCount < 60) {
                    Thread.sleep(1000);
                    waitCount++;
                    if (waitCount % 5 == 0) {
                        logger.info("⏳ 仍在等待连接... (已等待{}秒)", waitCount);
                    }
                }
                
                if (!reconnectableClient.isConnected()) {
                    logger.error("❌ 等待超时（60秒），客户端未能连接到服务器");
                    logger.info("💡 提示：请确保 ChatMcpServer 已在 8081 端口启动, 然后再手动运行此测试");
                    return;
                }
            }
            
            logger.info("✅ 客户端已连接到服务器，开始执行测试");

            // 3. 执行各种测试
//            runPingTest(reconnectableClient.getClient());
//            runGetAllConversationsTest(reconnectableClient.getClient());
//            runGetConversationByIdTest(reconnectableClient.getClient());
//            runGetMessagesByConversationIdTest(reconnectableClient.getClient());
//            runSendMessageTest(reconnectableClient.getClient());
//            runErrorTest(reconnectableClient.getClient());
            
            // 测试广播功能和定向消息
            logger.info("\n🎯 开始测试消息接收功能...");
            logger.info("💡 提示：客户端现在会实时接收服务器推送的消息");
            logger.info("==================================================\n");
            
//            runBroadcastTest(reconnectableClient.getClient());
//            runGetConnectedClientsTest(reconnectableClient.getClient());
            runSendToClientTest(reconnectableClient.getClient());

            logger.info("✅ 所有聊天服务测试完成");
            logger.info("💡 客户端正在运行中，按 Ctrl+C 或点击停止按钮关闭");
            
            // 保存主线程引用，以便 ShutdownHook 可以中断它
            Thread mainThread = Thread.currentThread();
            
            // 添加JVM关闭钩子
            Thread shutdownHook = new Thread(() -> {
                logger.info("🛑 检测到程序关闭信号...");
                // 中断主线程，让它能够退出 sleep
                mainThread.interrupt();
                // 关闭客户端
                reconnectableClient.shutdown();
            });
            Runtime.getRuntime().addShutdownHook(shutdownHook);

            // 让主线程持续运行，观察心跳和重连效果
            // 使用简单的 sleep 循环，这样可以响应中断
            try {
                while (!Thread.currentThread().isInterrupted()) {
                    Thread.sleep(1000);
                }
                logger.info("🛑 主线程收到中断信号，准备关闭...");
            } catch (InterruptedException e) {
                logger.info("🛑 主线程被中断，准备关闭...");
                Thread.currentThread().interrupt();
            }

        } catch (Exception e) {
            logger.error("❌ 测试失败", e);
        } finally {
            // 关闭客户端和心跳检测
            reconnectableClient.shutdown();
            logger.info("🛑 客户端已关闭");
            
            // 诊断：显示所有正在运行的非守护线程
            logger.info("🔍 检查剩余的非守护线程...");
            Thread.getAllStackTraces().keySet().stream()
                .filter(t -> !t.isDaemon() && t.isAlive())
                .forEach(t -> logger.info("  - 非守护线程: {} (状态: {})", t.getName(), t.getState()));
            
            // 给日志一点时间输出
            try {
                Thread.sleep(200);
            } catch (InterruptedException ignored) {
            }
            
            // 强制退出，确保进程能够终止
            logger.info("🛑 强制退出进程...");
            System.exit(0);
        }
    }

    /**
     * 测试0: Ping/Pong连接检测功能
     */
    private static void runPingTest(McpSyncClient client) {
        logger.info("\n🧪 测试0: Ping/Pong连接检测");
        logger.info("----------------------------");
        logger.info("💡 演示如何检测服务器连接状态");

        try {
            // 调用ping工具
            McpSchema.CallToolRequest request = new McpSchema.CallToolRequest("ping", Map.of());

            logger.info("🏓 发送ping请求...");
            logger.info("📝 注意观察实时日志推送 (这些信息通过SSE实时推送):");

            McpSchema.CallToolResult result = client.callTool(request);

            logger.info("✅ ping测试完成");
            logger.info("📄 ping响应: {}", result.content());

        } catch (Exception e) {
            logger.error("❌ ping测试失败", e);
        }
    }

    /**
     * 测试1: 获取所有聊天记录功能
     */
    private static void runGetAllConversationsTest(McpSyncClient client) {
        logger.info("\n🧪 测试1: 获取所有聊天记录");
        logger.info("----------------------------");
        logger.info("💡 演示如何获取所有聊天记录列表");

        try {
            // 初始化客户端
            client.initialize();
            logger.info("✅ 客户端初始化成功");

            // 调用获取所有聊天记录的工具
            McpSchema.CallToolRequest request = new McpSchema.CallToolRequest("get_all_conversations", Map.of());

            logger.info("📋 调用获取所有聊天记录...");
            logger.info("📝 注意观察实时日志推送 (这些信息通过SSE实时推送):");

            McpSchema.CallToolResult result = client.callTool(request);

            logger.info("✅ 获取聊天记录完成");
            logger.info("📄 聊天记录列表: {}", result.content());

        } catch (Exception e) {
            logger.error("❌ 获取聊天记录测试失败", e);
        }
    }

    /**
     * 测试2: 根据ID获取聊天记录功能
     */
    private static void runGetConversationByIdTest(McpSyncClient client) {
        logger.info("\n🧪 测试2: 根据ID获取聊天记录");
        logger.info("----------------------------");
        logger.info("💡 演示如何获取特定聊天记录的详细信息");

        try {
            // 调用根据ID获取聊天记录的工具
            McpSchema.CallToolRequest request = new McpSchema.CallToolRequest("get_conversation_by_id",
                    Map.of("conversation_id", "1"));

            logger.info("🔍 调用根据ID获取聊天记录...");
            logger.info("📝 注意观察实时日志推送 (这些信息通过SSE实时推送):");

            McpSchema.CallToolResult result = client.callTool(request);

            logger.info("✅ 获取聊天记录详情完成");
            logger.info("📄 聊天记录详情: {}", result.content());

        } catch (Exception e) {
            logger.error("❌ 获取聊天记录详情测试失败", e);
        }
    }

    /**
     * 测试3: 根据会话ID获取消息列表功能
     */
    private static void runGetMessagesByConversationIdTest(McpSyncClient client) {
        logger.info("\n🧪 测试3: 根据会话ID获取消息列表");
        logger.info("----------------------------");
        logger.info("💡 演示如何获取指定会话的消息列表");

        try {
            // 调用根据会话ID获取消息列表的工具
            McpSchema.CallToolRequest request = new McpSchema.CallToolRequest("get_messages_by_conversation_id",
                    Map.of("conversation_id", "1"));

            logger.info("📋 调用根据会话ID获取消息列表...");
            logger.info("📝 注意观察实时日志推送 (这些信息通过SSE实时推送):");

            McpSchema.CallToolResult result = client.callTool(request);

            logger.info("✅ 获取消息列表完成");
            logger.info("📄 消息列表: {}", result.content());

        } catch (Exception e) {
            logger.error("❌ 获取消息列表测试失败", e);
        }
    }

    /**
     * 测试4: 发送消息功能
     */
    private static void runSendMessageTest(McpSyncClient client) {
        logger.info("\n🧪 测试4: 发送消息");
        logger.info("----------------------------");
        logger.info("💡 演示如何发送消息到指定聊天记录");

        try {
            // 调用发送消息的工具
            McpSchema.CallToolRequest request = new McpSchema.CallToolRequest("send_message",
                    Map.of(
                            "conversation_id", "1",
                            "sender", "测试用户?",
                            "content", "这是一条测试消息，通过MCP客户端发送！"
                    ));

            logger.info("📤 调用发送消息...");
            logger.info("📝 注意观察实时日志推送 (这些信息通过SSE实时推送):");

            McpSchema.CallToolResult result = client.callTool(request);

            logger.info("✅ 发送消息完成");
            logger.info("📄 发送结果: {}", result.content());

        } catch (Exception e) {
            logger.error("❌ 发送消息测试失败", e);
        }
    }

    /**
     * 测试5: 错误处理功能
     */
    private static void runErrorTest(McpSyncClient client) {
        logger.info("\n🧪 测试5: 错误处理");
        logger.info("----------------------------");
        logger.info("💡 演示如何处理不存在的聊天记录ID");

        try {
            // 调用发送消息到不存在的聊天记录
            McpSchema.CallToolRequest request = new McpSchema.CallToolRequest("send_message",
                    Map.of(
                            "conversation_id", "999", // 不存在的ID
                            "sender", "测试用户",
                            "content", "这条消息应该失败"
                    ));

            logger.info("📤 调用发送消息到不存在的聊天记录...");
            logger.info("📝 注意观察错误处理:");

            McpSchema.CallToolResult result = client.callTool(request);

            if (result.isError()) {
                logger.info("✅ 错误处理正确 - 成功捕获了错误");
                logger.info("📄 错误信息: {}", result.content());
            } else {
                logger.warn("⚠️ 预期应该返回错误，但返回了成功结果");
            }

        } catch (Exception e) {
            logger.error("❌ 错误处理测试失败", e);
        }
    }

    /**
     * 测试6: 广播消息功能
     */
    private static void runBroadcastTest(McpSyncClient client) {
        logger.info("\n🧪 测试6: 广播消息功能");
        logger.info("----------------------------");
        logger.info("💡 演示如何向所有客户端广播消息");

        try {
            // 调用广播消息的工具
            McpSchema.CallToolRequest request = new McpSchema.CallToolRequest("broadcast_message",
                    Map.of(
                            "title", "系统通知",
                            "message", "这是一条测试广播消息，所有连接的客户端都会收到！"
                    ));

            logger.info("📢 调用广播消息...");
            logger.info("📝 注意：如果有多个客户端连接，它们都会收到这条消息");

            McpSchema.CallToolResult result = client.callTool(request);

            logger.info("✅ 广播消息发送完成");
            logger.info("📄 发送结果: {}", result.content());
            
            // 等待一下以便观察广播通知的接收
            Thread.sleep(2000);

        } catch (Exception e) {
            logger.error("❌ 广播消息测试失败", e);
        }
    }

    /**
     * 测试7: 获取在线客户端列表
     */
    private static void runGetConnectedClientsTest(McpSyncClient client) {
        logger.info("\n🧪 测试7: 获取在线客户端列表");
        logger.info("----------------------------");
        logger.info("💡 演示如何获取所有在线客户端的会话ID");

        try {
            // 调用获取在线客户端列表的工具
            McpSchema.CallToolRequest request = new McpSchema.CallToolRequest("get_connected_clients", Map.of());

            logger.info("📋 调用获取在线客户端列表...");

            McpSchema.CallToolResult result = client.callTool(request);

            logger.info("✅ 获取在线客户端列表完成");
            logger.info("📄 客户端列表: {}", result.content());

        } catch (Exception e) {
            logger.error("❌ 获取在线客户端列表测试失败", e);
        }
    }

    /**
     * 测试8: 向指定客户端推送消息
     */
    private static void runSendToClientTest(McpSyncClient client) {
        logger.info("\n🧪 测试8: 向指定客户端推送消息");
        logger.info("----------------------------");
        logger.info("💡 演示如何向指定的客户端推送消息");

        try {
            // 首先获取在线客户端列表
            McpSchema.CallToolRequest getClientsRequest = new McpSchema.CallToolRequest("get_connected_clients", Map.of());
            McpSchema.CallToolResult getClientsResult = client.callTool(getClientsRequest);
            
            logger.info("📋 当前在线客户端: {}", getClientsResult.content());
            
            // 解析在线客户端列表
            String contentStr = getClientsResult.content().toString();
            
            // 从结果中提取sessionId列表
            if (contentStr.contains("sessionId")) {
                logger.info("💡 从在线客户端列表中选择目标客户端");
                

                // 场景3: 测试向不存在的客户端发送消息（错误处理）
//                testSendToNonExistentClient(client);
                
                // 场景4: 向所有在线客户端发送消息
                testSendToAllClients(client);
                
            } else {
                logger.warn("⚠️  没有找到其他在线客户端");
                logger.info("💡 提示：可以启动多个客户端实例来测试客户端间消息推送");
            }

        } catch (Exception e) {
            logger.error("❌ 向指定客户端推送消息测试失败", e);
        }
    }
    

    /**
     * 测试场景2：向其他客户端发送消息
     */
    private void testSendToOtherClient(McpSyncClient client, String targetSessionId) {
        try {
            logger.info("\n📍 场景2: 向其他客户端发送消息");
            logger.info("目标客户端: {}", targetSessionId);
            
            McpSchema.CallToolRequest sendRequest = new McpSchema.CallToolRequest("send_to_client",
                    Map.of(
                            "session_id", targetSessionId,
                            "title", "点对点消息",
                            "message", "你好！这是来自 " + CUSTOM_SESSION_ID + " 的消息"
                    ));

            logger.info("📤 向客户端 {} 发送消息...", targetSessionId);
            McpSchema.CallToolResult sendResult = client.callTool(sendRequest);

            if (sendResult.isError()) {
                logger.warn("⚠️  目标客户端 {} 可能不在线", targetSessionId);
                logger.info("📄 错误信息: {}", sendResult.content());
            } else {
                logger.info("✅ 发送成功");
                logger.info("📄 发送结果: {}", sendResult.content());
            }
            
            Thread.sleep(1000);
            
        } catch (Exception e) {
            logger.error("❌ 场景2测试失败", e);
        }
    }
    
    /**
     * 测试场景3：向不存在的客户端发送消息（错误处理测试）
     */
    private static void testSendToNonExistentClient(McpSyncClient client) {
        try {
            logger.info("\n📍 场景3: 向不存在的客户端发送消息（错误处理测试）");
            String nonExistentSessionId = "non-existent-client-999";
            logger.info("目标客户端: {}", nonExistentSessionId);
            
            McpSchema.CallToolRequest sendRequest = new McpSchema.CallToolRequest("send_to_client",
                    Map.of(
                            "session_id", nonExistentSessionId,
                            "title", "测试消息",
                            "message", "这条消息应该无法送达"
                    ));

            logger.info("📤 向不存在的客户端 {} 发送消息...", nonExistentSessionId);
            McpSchema.CallToolResult sendResult = client.callTool(sendRequest);

            if (sendResult.isError()) {
                logger.info("✅ 错误处理正确 - 成功捕获了不存在的客户端错误");
                logger.info("📄 错误信息: {}", sendResult.content());
            } else {
                logger.warn("⚠️  预期应该返回错误，但返回了成功结果");
                logger.info("📄 结果: {}", sendResult.content());
            }
            
            Thread.sleep(1000);
            
        } catch (Exception e) {
            logger.error("❌ 场景3测试失败", e);
        }
    }
    
    /**
     * 测试场景4：向所有在线客户端发送消息（批量推送）
     */
    private static void testSendToAllClients(McpSyncClient client) {
        try {
            logger.info("\n📍 场景4: 向所有在线客户端发送消息（批量推送）");
            logger.info("💡 这个场景演示如何使用 send_to_client 向所有用户发送消息");
            
            // 获取在线客户端列表
            McpSchema.CallToolRequest getClientsRequest = new McpSchema.CallToolRequest("get_connected_clients", Map.of());
            McpSchema.CallToolResult getClientsResult = client.callTool(getClientsRequest);
            
            // 从content中提取实际的文本内容
            // [TextContent[annotations=null, text={"sessionIds": ["client-00001-1760238687227","client-00001-1760238708225"]}, meta=null]]
            String clientsData = null;
            if (!getClientsResult.content().isEmpty() && getClientsResult.content().get(0) instanceof McpSchema.TextContent) {
                clientsData = ((McpSchema.TextContent) getClientsResult.content().get(0)).text();
            }
            
            if (clientsData == null || clientsData.isEmpty()) {
                logger.warn("⚠️ 未获取到客户端列表数据");
                return;
            }
            
            logger.info("📋 获取到的客户端列表数据: {}", clientsData);
            
            try {
                // 尝试从content中提取sessionId列表
                // 根据实际返回格式调整解析逻辑
                java.util.List<String> sessionIds = extractSessionIds(clientsData);

                // 如果JSON解析失败，尝试文本匹配
                if (sessionIds.isEmpty()) {
                    logger.info("💡 JSON解析未找到客户端...");
                    throw new Exception("无法解析客户端列表");
                }
                
                if (sessionIds.isEmpty()) {
                    logger.warn("⚠️  没有找到在线客户端");
                    logger.warn("📋 原始数据: {}", clientsData);
                    return;
                }
                
                logger.info("✅ 发现 {} 个在线客户端: {}", sessionIds.size(), sessionIds);
                logger.info("📤 开始向所有客户端发送消息...\n");
                
                int successCount = 0;
                int failCount = 0;
                
                // 遍历所有客户端，逐个发送消息
                for (String sessionId : sessionIds) {
                    try {
                        logger.info("  ➤ 正在向客户端 {} 发送消息...", sessionId);
                        
                        McpSchema.CallToolRequest sendRequest = new McpSchema.CallToolRequest("send_to_client",
                                Map.of(
                                        "session_id", sessionId,
                                        "title", "群发通知",
                                        "message", "这是一条发送给所有在线用户的群发消息！客户端发送时间: " + new java.util.Date()
                                ));
                        
                        McpSchema.CallToolResult sendResult = client.callTool(sendRequest);
                        
                        if (sendResult.isError()) {
                            logger.warn("    ✗ 发送失败: {}", sendResult.content());
                            failCount++;
                        } else {
                            logger.info("    ✓ 发送成功");
                            successCount++;
                        }
                        
                        // 避免发送太快，稍微延迟一下
                        Thread.sleep(500);
                        
                    } catch (Exception e) {
                        logger.error("    ✗ 发送异常: {}", e.getMessage());
                        failCount++;
                    }
                }
                
                logger.info("\n📊 批量发送统计:");
                logger.info("  ✅ 成功: {} 个", successCount);
                logger.info("  ❌ 失败: {} 个", failCount);
                logger.info("  📈 成功率: {}/{} ({}%)", 
                        successCount, sessionIds.size(), 
                        sessionIds.isEmpty() ? 0 : (successCount * 100 / sessionIds.size()));
                
            } catch (Exception e) {
                logger.error("❌ 发送消息过程中出现异常: {}", e.getMessage(), e);
            }
            
            Thread.sleep(2000);
            
        } catch (Exception e) {
            logger.error("❌ 场景4测试失败", e);
        }
    }
    
    /**
     * 从JSON或文本中提取sessionId列表
     */
    private static java.util.List<String> extractSessionIds(String data) {
        java.util.List<String> sessionIds = new java.util.ArrayList<>();
        
        try {
            // 尝试解析为JSON
            McpJsonMapper jsonMapper = McpJsonMapper.getDefault();
            
            // 尝试多种可能的JSON格式
            if (data.trim().startsWith("[")) {
                // 数组格式: ["id1", "id2", ...]
                @SuppressWarnings("unchecked")
                java.util.List<Object> list = jsonMapper.readValue(data, java.util.List.class);
                for (Object obj : list) {
                    if (obj instanceof String) {
                        sessionIds.add((String) obj);
                    } else if (obj instanceof Map) {
                        @SuppressWarnings("unchecked")
                        Map<String, Object> map = (Map<String, Object>) obj;
                        if (map.containsKey("sessionId")) {
                            sessionIds.add(map.get("sessionId").toString());
                        }
                    }
                }
            } else if (data.trim().startsWith("{")) {
                // 对象格式: {"sessionIds": [...]}
                @SuppressWarnings("unchecked")
                Map<String, Object> map = jsonMapper.readValue(data, Map.class);
                if (map.containsKey("sessionIds")) {
                    @SuppressWarnings("unchecked")
                    java.util.List<String> ids = (java.util.List<String>) map.get("sessionIds");
                    sessionIds.addAll(ids);
                }
            }
        } catch (Exception e) {
            // JSON解析失败，返回空列表
        }
        
        return sessionIds;
    }
    
    /**
     * 处理自定义通知（从日志消息中解析）
     */
    private static void handleCustomNotification(McpSchema.LoggingMessageNotification notification) {
        try {
            // 日志数据是JSON字符串，需要先解析
            String dataStr = notification.data();
            if (dataStr == null || dataStr.isEmpty()) {
                return;
            }
            
            // 解析JSON字符串为Map
            McpJsonMapper jsonMapper = McpJsonMapper.getDefault();
            @SuppressWarnings("unchecked")
            Map<String, Object> dataMap = jsonMapper.readValue(dataStr, Map.class);
            
            String method = (String) dataMap.get("method");
            Object params = dataMap.get("params");
            
            // 根据方法名分发处理
            if ("notifications/message".equals(method)) {
                handlePersonalMessage(params);
            } else if ("notifications/broadcast".equals(method)) {
                handleBroadcastMessage(params);
            } else {
                logger.warn("⚠️ 收到未知的自定义通知: {}", method);
            }
        } catch (Exception e) {
            logger.error("❌ 处理自定义通知失败", e);
        }
    }
    
    /**
     * 处理个人消息
     */
    private static void handlePersonalMessage(Object params) {
        if (params instanceof Map) {
            @SuppressWarnings("unchecked")
            Map<String, Object> messageMap = (Map<String, Object>) params;
            String sessionId = (String) messageMap.get("sessionId");
            String title = (String) messageMap.get("title");
            String message = (String) messageMap.get("message");
            Object timestampObj = messageMap.get("timestamp");
            Long timestamp = timestampObj != null ? ((Number) timestampObj).longValue() : System.currentTimeMillis();
            
            logger.info("╔════════════════════════════════════════════╗");
            logger.info("║       📬 收到个人消息！                       ║");
            logger.info("╠════════════════════════════════════════════╣");
            logger.info("║ 📣 sessionId: {}", sessionId);
            logger.info("║ 📣 标题: {}", title);
            logger.info("║ 📝 内容: {}", message);
            logger.info("║ ⏰ 服务器推送时间: {}", new java.util.Date(timestamp));
            logger.info("╚════════════════════════════════════════════╝\n");
        }
    }
    
    /**
     * 处理广播消息
     */
    private static void handleBroadcastMessage(Object params) {
        if (params instanceof Map) {
            @SuppressWarnings("unchecked")
            Map<String, Object> messageMap = (Map<String, Object>) params;
            String title = (String) messageMap.get("title");
            String message = (String) messageMap.get("message");
            Object timestampObj = messageMap.get("timestamp");
            Long timestamp = timestampObj != null ? ((Number) timestampObj).longValue() : System.currentTimeMillis();
            
            logger.info("╔════════════════════════════════════════════╗");
            logger.info("║       📢 收到广播消息！                       ║");
            logger.info("╠════════════════════════════════════════════╣");
            logger.info("║ 📣 标题: {}", title);
            logger.info("║ 📝 内容: {}", message);
            logger.info("║ ⏰ 时间: {}", new java.util.Date(timestamp));
            logger.info("╚════════════════════════════════════════════╝\n");
        }
    }

    /**
     * 可重连的MCP客户端管理器
     * <p>
     * 功能：
     * 1. 自动创建和初始化MCP客户端
     * 2. 定期执行ping检测连接状态
     * 3. 检测到连接断开时自动重连
     * 4. 支持优雅关闭
     */
    class ReconnectableMcpClient {
        private static final Logger logger = LoggerFactory.getLogger(ReconnectableMcpClient.class);
        
        private final String serverUrl;
        private volatile McpSyncClient client;
        private volatile McpClientTransport transport;
        
        private final ScheduledExecutorService scheduler;
        private final AtomicBoolean isRunning;
        private final AtomicBoolean isConnected;
        private final AtomicInteger reconnectAttempts;
        
        private ScheduledFuture<?> pingTask;
        
        public ReconnectableMcpClient(String serverUrl) {
            this.serverUrl = serverUrl;
            // 使用守护线程池，避免阻止JVM退出
            this.scheduler = Executors.newScheduledThreadPool(2, r -> {
                Thread thread = new Thread(r);
                thread.setDaemon(true);  // 设置为守护线程
                thread.setName("MCP-Client-Scheduler");
                return thread;
            });
            this.isRunning = new AtomicBoolean(false);
            this.isConnected = new AtomicBoolean(false);
            this.reconnectAttempts = new AtomicInteger(0);
        }
        
        /**
         * 启动客户端并开始心跳检测
         */
        public void start() {
            if (isRunning.compareAndSet(false, true)) {
                logger.info("🔌 正在启动MCP客户端...");

                connect();
                // 无论初次连接是否成功，都启动心跳检测
                startPingTask();
                logger.info("✅ MCP客户端启动完成（心跳检测已开启）");
            }
        }
        
        /**
         * 创建连接
         */
        private void connect() {
            try {
                logger.info("🔗 正在连接到服务器: {}", serverUrl);
                
                // 1. 创建HTTP SSE传输层
                // 如果指定了自定义会话ID，则通过URL参数传递
                String sseEndpoint = "/sse";
                if (CUSTOM_SESSION_ID != null && !CUSTOM_SESSION_ID.isEmpty()) {
                    sseEndpoint = "/sse?sessionId=" + CUSTOM_SESSION_ID;
                    logger.info("🔑 使用自定义会话ID: {}", CUSTOM_SESSION_ID);
                }
                
                transport = HttpClientSseClientTransport.builder(serverUrl)
                        .sseEndpoint(sseEndpoint)
                        .build();
                
                logger.debug("📡 HTTP SSE传输层创建成功");
                
                // 2. 创建同步MCP客户端
                // 使用日志消息机制接收自定义通知
                client = McpClient.sync(transport)
                        .requestTimeout(Duration.ofSeconds(30))
                        .initializationTimeout(Duration.ofSeconds(10))
                        .capabilities(McpSchema.ClientCapabilities.builder().roots(true).sampling().build())
                        .clientInfo(new McpSchema.Implementation("chat-mcp-client", "1.0.0"))
                        .loggingConsumer(notification -> {
                            // 检查是否是自定义通知（通过特殊的logger名称识别）
                            if ("CustomNotification".equals(notification.logger())) {
                                handleCustomNotification(notification);
                            } else {
                                // 普通日志消息
                                logger.info("📝 [实时日志] [{}] {}: {}", 
                                        notification.level(), notification.logger(), notification.data());
                            }
                        })
                        .progressConsumer(notification -> {
                            logger.info("📊 [进度更新] {}: {}/{} - {}", 
                                    notification.progressToken(), notification.progress(),
                                    notification.total(), notification.message());
                        })
                        .build();
                
                logger.debug("🔧 MCP客户端创建成功，开始初始化...");
                
                // 3. 初始化客户端
                client.initialize();
                
                isConnected.set(true);
                reconnectAttempts.set(0);
                logger.info("✅ 客户端连接并初始化成功");
                
            } catch (Exception e) {
                isConnected.set(false);
                logger.error("❌ 连接失败: {} - {}", e.getClass().getSimpleName(), e.getMessage());
            }
        }
        
        /**
         * 启动ping心跳检测任务
         */
        private void startPingTask() {
            pingTask = scheduler.scheduleAtFixedRate(() -> {
                if (!isRunning.get()) {
                    return;
                }
                
                try {
                    // 执行ping检测
                    boolean pingSuccess = ping();
                    
                    if (!pingSuccess) {
                        logger.warn("⚠️ Ping失败，连接可能已断开");
                        handleConnectionLoss();
                    } else {
                        // ping成功，重置重连计数
                        if (reconnectAttempts.get() > 0) {
                            reconnectAttempts.set(0);
                        }
                    }
                } catch (Exception e) {
                    logger.warn("⚠️ Ping检测异常: {}", e.getMessage());
                    handleConnectionLoss();
                }
            }, PING_INTERVAL_SECONDS, PING_INTERVAL_SECONDS, TimeUnit.SECONDS);
            
            logger.info("💓 心跳检测已启动（间隔: {}秒）", PING_INTERVAL_SECONDS);
        }
        
        /**
         * 执行ping操作
         */
        private boolean ping() {
            if (client == null) {
                return false;
            }
            
            try {
                logger.debug("🏓 发送ping...");
                
                // 创建一个Future来实现超时控制
                // 使用守护线程，避免阻止JVM退出
                ExecutorService executor = Executors.newSingleThreadExecutor(r -> {
                    Thread thread = new Thread(r);
                    thread.setDaemon(true);
                    thread.setName("MCP-Ping-Executor");
                    return thread;
                });
                Future<Boolean> future = executor.submit(() -> {
                    try {
                        McpSchema.CallToolRequest request = new McpSchema.CallToolRequest("ping", Map.of());
                        McpSchema.CallToolResult result = client.callTool(request);
                        
                        if (result != null && !result.isError()) {
                            logger.debug("✅ Pong收到");
                            return true;
                        }
                        return false;
                    } catch (Exception e) {
                        logger.debug("❌ Ping异常: {}", e.getMessage());
                        return false;
                    }
                });
                
                try {
                    Boolean result = future.get(PING_TIMEOUT_SECONDS, TimeUnit.SECONDS);
                    executor.shutdown();
                    return result != null && result;
                } catch (TimeoutException e) {
                    logger.warn("⏰ Ping超时（{}秒）", PING_TIMEOUT_SECONDS);
                    future.cancel(true);
                    executor.shutdownNow();
                    return false;
                }
                
            } catch (Exception e) {
                logger.debug("❌ Ping执行失败: {}", e.getMessage());
                return false;
            }
        }
        
        /**
         * 处理连接丢失
         */
        private void handleConnectionLoss() {
            // 确保连接状态被标记为断开
            boolean wasConnected = isConnected.getAndSet(false);
            
            int attempts = reconnectAttempts.get();
            
            // 如果已经在重连中（重连次数 > 0）且连接已经断开，避免重复触发重连
            if (!wasConnected && attempts > 0) {
                logger.debug("⏳ 重连已在进行中（第{}次尝试），跳过", attempts);
                return;
            }
            
            final int currentAttempt = reconnectAttempts.incrementAndGet();
            
            if (currentAttempt > MAX_RECONNECT_ATTEMPTS) {
                logger.error("❌ 已达到最大重连次数({}次)，停止重连", MAX_RECONNECT_ATTEMPTS);
                isRunning.set(false);
                return;
            }
            
            logger.warn("🔄 检测到连接断开，准备重连（第{}次尝试，将在{}秒后执行）", currentAttempt, RECONNECT_DELAY_SECONDS);
            
            // 在后台线程执行重连
            scheduler.schedule(() -> {
                logger.info("⏰ 重连任务开始执行（第{}次尝试）", currentAttempt);
                try {
                    reconnect();
                    logger.info("✅ 第{}次重连成功！", currentAttempt);
                } catch (Exception e) {
                    logger.error("❌ 第{}次重连失败: {} - {}", currentAttempt, e.getClass().getSimpleName(), e.getMessage());
                    // 重连失败，继续尝试下一次（通过下一次ping失败触发）
                    isConnected.set(false);
                }
            }, RECONNECT_DELAY_SECONDS, TimeUnit.SECONDS);
        }
        
        /**
         * 执行重连
         */
        private void reconnect() throws Exception {
            logger.info("🔄 开始重连...");
            
            // 1. 关闭旧连接
            closeClientQuietly();
            
            // 2. 等待一小段时间
            Thread.sleep(1000);
            
            // 3. 重新连接
            connect();
        }
        
        /**
         * 安静地关闭客户端（忽略错误）
         */
        private void closeClientQuietly() {
            logger.debug("🔌 关闭旧的客户端连接...");
            
            // 先关闭客户端
            if (client != null) {
                try {
                    client.close();
                    logger.debug("✓ 客户端已关闭");
                } catch (Exception e) {
                    logger.debug("关闭客户端时出错（忽略）: {}", e.getMessage());
                }
            }
            
            // 再关闭 transport
            if (transport != null) {
                try {
                    transport.close();
                    logger.debug("✓ Transport已关闭");
                } catch (Exception e) {
                    logger.debug("关闭transport时出错（忽略）: {}", e.getMessage());
                }
            }
            
            // 清空引用
            client = null;
            transport = null;
        }
        
        /**
         * 获取当前客户端实例
         */
        public McpSyncClient getClient() {
            if (!isConnected.get() || client == null) {
                logger.error("❌ 客户端未连接，无法获取实例");
            }
            return client;
        }
        
        /**
         * 检查是否已连接
         */
        public boolean isConnected() {
            return isConnected.get();
        }
        
        /**
         * 关闭客户端和心跳检测
         */
        public void shutdown() {
            logger.info("🛑 正在关闭MCP客户端...");
            
            isRunning.set(false);
            isConnected.set(false);
            
            // 先尝试主动通知服务器断开连接
            notifyServerDisconnect();
            
            // 停止ping任务
            if (pingTask != null) {
                pingTask.cancel(true);
            }
            
            // 关闭客户端
            closeClientQuietly();
            
            // 关闭调度器
            scheduler.shutdownNow();
            
            // 由于使用守护线程，不需要等待，直接退出即可
            // JVM 会自动终止守护线程
            
            logger.info("✅ MCP客户端已关闭");
        }
        
        /**
         * 通知服务器客户端即将断开连接
         */
        private void notifyServerDisconnect() {
            if (client == null) {
                return;
            }
            
            try {
                logger.info("📤 通知服务器客户端即将断开连接...");
                
                // 创建一个Future来实现超时控制，避免阻塞关闭流程
                // 使用守护线程，避免阻止JVM退出
                ExecutorService executor = Executors.newSingleThreadExecutor(r -> {
                    Thread thread = new Thread(r);
                    thread.setDaemon(true);
                    thread.setName("MCP-Disconnect-Notifier");
                    return thread;
                });
                Future<Void> future = executor.submit(() -> {
                    try {
                        McpSchema.CallToolRequest request = new McpSchema.CallToolRequest("disconnect", Map.of());
                        McpSchema.CallToolResult result = client.callTool(request);
                        
                        if (result != null && !result.isError()) {
                            logger.info("✅ 已成功通知服务器断开连接");
                        }
                    } catch (Exception e) {
                        logger.debug("通知服务器断开时出错（忽略）: {}", e.getMessage());
                    }
                    return null;
                });
                
                try {
                    // 最多等待2秒，避免阻塞关闭流程
                    future.get(2, TimeUnit.SECONDS);
                } catch (TimeoutException e) {
                    logger.debug("通知服务器断开超时（忽略）");
                    future.cancel(true);
                } finally {
                    executor.shutdownNow();
                }
                
            } catch (Exception e) {
                // 忽略所有异常，确保关闭流程能继续
                logger.debug("通知服务器断开失败（忽略）: {}", e.getMessage());
            }
        }
    }
}
